/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.jms;

import java.util.Vector;

import javax.jms.ConnectionConsumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ServerSessionPool;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;

/**
   This class implements the TopicConnection behavior outlined in the JMS 
   specification.

   @author Shrideep Pallickara
   $Revision$
   $Date$
*/ 

public class JmsTopicConnection extends JmsConnection 
  implements TopicConnection, JmsDebugFlags {

  /** These variable indicate the state of the connection */
  private boolean stopped = true; /** Connection is in the start/stop state */ 
  private boolean closed = false; /** Connection is closed? */
  
   /** This vector maintains a list of the sessions that are registered to
      this connection. */
  private Vector sessions;

  private Vector messagesDuringStop;

 
  private NBJmsInterconnectionBridge nbBridge;
  

  
  public JmsTopicConnection(NBJmsInterconnectionBridge nbBridge) {
    this.nbBridge = nbBridge;
    sessions = new Vector();
    messagesDuringStop = new Vector();
    nbBridge.setJmsTopicConnection(this);
  }


  public NBJmsInterconnectionBridge getNbJmsInterconnectionBridge() {
    return nbBridge;
  }

  public TopicSession createTopicSession(boolean transacted,
					 int acknowledgeMode)
    throws JMSException {
    
    JmsTopicSession jmsTopicSession = 
      new JmsTopicSession(this, transacted, acknowledgeMode);
    addToListOfManagedSessions(jmsTopicSession);
    return jmsTopicSession;
  }


  public ConnectionConsumer 
  createConnectionConsumer(Topic topic,
			   String messageSelector,
			   ServerSessionPool sessionPool,
			   int maxMessages)
    throws JMSException {
    
    /** JMS-NI */
    return null;
  }
  
  
  public ConnectionConsumer 
  createDurableConnectionConsumer(Topic topic,
				  java.lang.String subscriptionName,
				  java.lang.String messageSelector,
				  ServerSessionPool sessionPool,
				  int maxMessages)
    throws JMSException {
    /** JMS-NI */
    return null;
  }

  
  

    /** 
      Starts (or restarts) a connection's delivery of incoming messages.
      A call to <CODE>start</CODE> on a connection that has already been
      started is ignored.
      @exception JMSException if the JMS provider fails to start
      message delivery due to some internal error.
      @see javax.jms.Connection#stop
  */
  public void start() throws JMSException {
    if (closed)
      throw 
	new javax.jms.IllegalStateException("Peforming the start operation on"
					    + " a closed Connection");
    if (!stopped) {
      /* No need to start any sessions, since they are already started */
      return;
    }
    
    /*********** Start every session that is registered to this connection */
    for (int i=0; i < sessions.size(); i++) {
      ((JmsTopicSession)sessions.elementAt(i)).startSession();
    }
    
    if (messagesDuringStop.size() > 0) {
      System.out.println("JmsTopicConnection proceeding to release messages" +
			 " that were cached during the time connection was " +
			 "closed");

    }

    /** Release the set of messages received during stop */
    for (int i=0; i< messagesDuringStop.size(); i++) {
      JmsEvent _jmsEvent = (JmsEvent) messagesDuringStop.elementAt(i);
      routeEventToSessions(_jmsEvent);
    }
    stopped = false;
    messagesDuringStop.removeAllElements();
  }
  
  /** Temporarily stops a connection's delivery of incoming messages.
      Delivery can be restarted using the connection's <CODE>start</CODE>
      method. When the connection is stopped, delivery to all the connection's
      message consumers is inhibited: synchronous receives block, and messages
      are not delivered to message listeners.
      
      @exception JMSException if the JMS provider fails to stop message 
      delivery due to some internal error.
      @see javax.jms.Connection#start
  */
  public void stop() throws JMSException {
    if (closed)
      throw 
	new javax.jms.IllegalStateException("Peforming the stop operation on" +
					    " a closed Connection");
    if (stopped) {
      /* No need to stop any sessions, since they are already stopped */
      return;
    }
    
    /**** Stop every session that is registered to this connection ****/
    for (int i=0; i < sessions.size(); i++) {
      ((JmsTopicSession)sessions.elementAt(i)).stopSession();
    }
    stopped = true;
  }
  
  
  /** Closes the connection.
      @exception JMSException if the JMS provider fails to close the
      connection due to some internal error. For example, a failure to release
      resources or to close a socket connection can cause this exception to 
      be thrown.
  */
  public  void close() throws JMSException {
    if (closed) {
      /** No need to do anything further, simply return */
      return;
    }
    
    if (!stopped) {
      /** Stop all the sessions prior to closing them */
      for (int i=0; i < sessions.size(); i++) {
	((JmsTopicSession)sessions.elementAt(i)).stopSession();
      }
      stopped = true;
    }/*end-if */
    
    /** Proceed to close all sessions */
    for (int i=0; i < sessions.size(); i++) {
      ((JmsTopicSession)sessions.elementAt(i)).close();
    }
    /** Clear the set of managed sessions */
    sessions.clear();

    nbBridge.closeConnection();
    
    closed = true;
  }
  
  
  public void addToListOfManagedSessions(JmsTopicSession session) {
    sessions.addElement(session);
  }
  
  
  /** This is invoked by the Interconnection Bridge */
  public void processDataReceived(byte[] data) {
    if (closed) {
      System.out.println("Connection has been closed, so messages " + 
			 "being discarded");
      return;
    }
    /** Proceed to de-serialize the data, among other things - decide which
	sessions the events must be routed to ... etc. The list of sessions 
	is contained in the super-class JmsConnection */
    JmsEvent jmsEventRecvd = new JmsEvent(data); 
    if (JmsTopicConnection_Debug) {
      try {
	String _jmsTopic = jmsEventRecvd.getTopic();
	Message _jmsMsg = jmsEventRecvd.getMessage();
	String _txtMsgInfo ="";
	if (_jmsMsg instanceof JmsTextMessage) {
	  _txtMsgInfo = "Text Message (" + 
	    ((JmsTextMessage)_jmsMsg).getText() + ")" ;
	}
	System.out.println("JmsTopicConnection: RECEIVED EVENT =>>  " +
			   " Topic " +  _jmsTopic + 
			   _txtMsgInfo);
      } catch (JMSException jmsEx) {
	System.out.println("Error printing RecvdEventInfo in TopicConnection" +
			   jmsEx);
      }
    }
    
    if (stopped) {
      String _jmsTopic = jmsEventRecvd.getTopic();
      System.out.println("Message has been received, however, since the " +
			 "connection is not started, delivery inhibited" + 
			 " on topic => " + _jmsTopic);
      messagesDuringStop.addElement(jmsEventRecvd);
      return;
    }
    
    routeEventToSessions(jmsEventRecvd);
  }
  
  private void routeEventToSessions(JmsEvent jmsEventRecvd) {
    /** There are usually multiple sessions setup on the same connection */
    if (JmsTopicConnection_Debug) {
      System.out.println("Trying to route event to subscribers ... ");
    }
    for (int i=0; i < sessions.size(); i++) {
      JmsTopicSession _topicSession = (JmsTopicSession) sessions.elementAt(i);
      _topicSession.routeEventToAppropriateSubscribers(jmsEventRecvd);
    }
    
    //Topic topicSentTo = jmsEventRecvd.getTopic();
    /** Based on the topic that it was sent to, route it to the appropriate 
	session. Sessions should maintain the list of topics that clients 
	have subscribed to.
    */
  }
}

